//package com.wutoon.etl.processor;
//
//import com.alibaba.fastjson.JSON;
//import com.alibaba.fastjson.JSONObject;
//import com.alibaba.fastjson.TypeReference;
//import lombok.AllArgsConstructor;
//import lombok.Getter;
//import lombok.NoArgsConstructor;
//import lombok.extern.slf4j.Slf4j;
//import okhttp3.*;
//import okhttp3.internal.sse.RealEventSource;
//import okhttp3.sse.EventSource;
//import okhttp3.sse.EventSourceListener;
//import org.apache.commons.lang3.ObjectUtils;
//import org.apache.commons.lang3.StringUtils;
//import org.springframework.http.HttpMethod;
//import org.springframework.stereotype.Component;
//import tech.powerjob.common.serialize.JsonUtils;
//import tech.powerjob.common.utils.CollectionUtils;
//import tech.powerjob.official.processors.constant.ProcessorConstant;
//import tech.powerjob.official.processors.util.CommonUtils;
//import tech.powerjob.official.processors.util.HttpClientHolder;
//import tech.powerjob.worker.common.ThreadLocalStore;
//import tech.powerjob.worker.core.processor.ProcessResult;
//import tech.powerjob.worker.core.processor.TaskContext;
//import tech.powerjob.worker.core.processor.TaskResult;
//import tech.powerjob.worker.core.processor.WorkflowContext;
//import tech.powerjob.worker.core.processor.sdk.MapReduceProcessor;
//import tech.powerjob.worker.enums.SseStage;
//import tech.powerjob.worker.log.OmsLogger;
//import tech.powerjob.worker.pojo.model.MapReduceShard;
//import tech.powerjob.worker.pojo.SseOutputInfo;
//import tech.powerjob.worker.pojo.model.HttpModel;
//import tech.powerjob.worker.pojo.model.MapReduceHttpModel;
//
//import javax.annotation.Nonnull;
//import java.io.IOException;
//import java.math.BigDecimal;
//import java.util.ArrayList;
//import java.util.List;
//import java.util.Optional;
//import java.util.concurrent.CountDownLatch;
//import java.util.concurrent.TimeUnit;
//import java.util.stream.Collectors;
//
///**
// * MapReduce + http处理器 使用mapreduce的方式调用业务方数据
// * 控制台参数：{"batchSize": 100, "batchNum": 2}
// *
// * @author tjq
// * @since 2020/4/17
// */
//@Slf4j
//@Component
//public class MapReduceHttpProcessor implements MapReduceProcessor {
//
//    @Override
//    public ProcessResult process(TaskContext context) throws Exception {
//        OmsLogger omsLogger = context.getOmsLogger();
//        ThreadLocalStore.setLocalData(ProcessorConstant.INSTANCE_ID, String.valueOf(context.getInstanceId()));
//        omsLogger.info("============== MapHttpProcessor#process start==============");
//        omsLogger.info("MapHttpProcessor[{}]->MapHttpProcessor main: taskContext:{}",
//                ThreadLocalStore.getLocalData(ProcessorConstant.INSTANCE_ID), JsonUtils.toJSONString(context));
//        try {
//
//            // 根据控制台参数获取MR批次及子任务大小
//            String jobParams = CommonUtils.parseParams(context);
//            // 替换时间变量
//            omsLogger.info("MapHttpProcessor[{}]->MapHttpProcessor main: 开始替换参数变量", context.getInstanceId());
//            jobParams = tech.powerjob.common.utils.CommonUtils.replaceVariable(jobParams);
//            HttpModel<MapReduceHttpModel> contextParam = Optional.of(jobParams)
//                    .map(param -> JSON.parseObject(param, new TypeReference<HttpModel<MapReduceHttpModel>>(){}))
//                    .orElse(new HttpModel<>());
//            // 校验该执行的必要参数
//            validateParam(omsLogger, contextParam);
//
//            // root task 负责分发任务, 根据参数中配置的分片参数分发子任务
//            if (isRootTask()) {
//                // 替换参数中的变量，支持时间变量 TODO
//                omsLogger.info("MapHttpProcessor[{}]->MapHttpProcessor main: 参数变量替换完毕", context.getInstanceId());
//
//                MapReduceHttpModel httpParam = contextParam.getBody();
//                Integer shardTotal = httpParam.getShardTotal();
//                omsLogger.info("MapHttpProcessor[{}]->MapHttpProcessor main: 主分片开始组装任务，分片总数:{}", context.getInstanceId(), shardTotal);
//
//                // 分发逻辑：1、按照分片总数，直接发送透传参数
//                List<MapReduceShard> subTasks = new ArrayList<>();
//                for (int i = 0; i < shardTotal; i ++) {
//                    subTasks.add(buildShard(context, httpParam, i));
//                }
//                omsLogger.info("MapHttpProcessor[{}]->MapHttpProcessor main: 主分片开始分发任务，分片总数:{}", context.getInstanceId(), shardTotal);
//                map(subTasks, httpParam.getTaskName());
//                omsLogger.info("MapHttpProcessor[{}]->MapHttpProcessor main: 主分片开始任务分发完毕，分片总数:{}", context.getInstanceId(), shardTotal);
//                return new ProcessResult(true, ProcessorConstant.ROOT_TASK_RESULT);
//            }
//            // 子任务处理逻辑
//            Object subTask = context.getSubTask();
//            omsLogger.info("MapHttpProcessor[{}]->MapHttpProcessor main: 子任务开始：{}", context.getInstanceId(), JSON.toJSONString(subTask));
//            if (subTask instanceof MapReduceShard) {
//                // 实际处理
//                return dispatchBusinessTask(contextParam, (MapReduceShard)subTask, omsLogger);
//            }
//            omsLogger.info("MapHttpProcessor[{}]->MapHttpProcessor main: 子任务未正常开启，未知异常：{}", context.getInstanceId(), JSON.toJSONString(subTask));
//            return new ProcessResult(false, "UNKNOWN BUG");
//        } catch (Exception e) {
//            omsLogger.error("MapHttpProcessor main: error, param:{}, cause:", JsonUtils.toJSONString(context), e);
//            return new ProcessResult(false, e.getMessage());
//        }
//    }
//
///**
// * 构建分片参数
// *
// * @param context TaskContext
// * @param httpParam MapReduceHttpModel
// * @param i int
// * @return MapReduceShard
// * @author 武兴云/72176468
// * @date 2024/1/5 11:15
// * @update_by 武兴云/72176468
// * @update_at 2024/1/5 11:15
// * @creed if you have doubt , please contact me !!!
// */
//private MapReduceShard buildShard(TaskContext context, MapReduceHttpModel httpParam, int i) {
//    // 组装基础信息
//    MapReduceShard shard = MapReduceShard.builder().tenantId(context.getAppName())
//            .workFlowName(context.getRealWorkFlowName())
//            .taskName(context.getRealTaskName())
//            .shardTotal(httpParam.getShardTotal())
//            .taskInstanceId(context.getInstanceId())
//            .wfInstanceId(Optional.ofNullable(context.getWorkflowContext()).map(WorkflowContext::getWfInstanceId).orElse(0L))
//            .shardNum(i)
//            .customParam(httpParam.getCustomParam())
//            .build();
//
//    // 组装分片信息，把时间区间按照各自分片号分配
//    // 如果分片数大于1，计算每个分片处理的时间范围
//    if (httpParam.getShardTotal() > BigDecimal.ONE.intValue()) {
//        long interval = httpParam.getEndTime() - httpParam.getStartTime();
//        long perMills = interval / httpParam.getShardTotal();
//        shard.setStartTime(httpParam.getStartTime() + i * perMills);
//        shard.setEndTime(httpParam.getStartTime() + (i + 1) * perMills);
//    } else {
//        shard.setStartTime(httpParam.getStartTime());
//        shard.setEndTime(httpParam.getEndTime());
//    }
//    return shard;
//}
//
//    /**
//     * 任务分发
//     *
//     * @param contextParam 任务参数
//     * @param subTask 子任务
//     * @param omsLogger 日志记录器
//     * @return result
//     * @author 武兴云/72176468
//     * @date 2024-01-03 15:45
//     * @update_by 武兴云/72176468
//     * @update_at 2024-01-03 15:45
//     * @creed if you have doubt , please contact me !!!
//     */
//    private ProcessResult dispatchBusinessTask(HttpModel<MapReduceHttpModel> contextParam, MapReduceShard subTask, OmsLogger omsLogger) throws IOException {
//        omsLogger.info("dispatchBusinessTask: 子任务{}开始执行：{}", subTask.getShardNum(), JSON.toJSONString(contextParam));
//        OkHttpClient client = HttpClientHolder.getClient(contextParam.getTimeout());
//        omsLogger.info("dispatchBusinessTask: 子任务{}获取httpClient完毕", subTask.getShardNum());
//        Request.Builder builder = new Request.Builder().url(contextParam.getUrl());
//        if (!contextParam.getHeaders().isEmpty()) {
//            contextParam.getHeaders().forEach((k, v) -> {
//                builder.addHeader(k, v);
//                omsLogger.info("dispatchBusinessTask：子任务：{} add header {}:{}", subTask.getShardNum(), k, v);
//            });
//            if (contextParam.getFluent()) {
//                builder.removeHeader("Accept");
//                builder.addHeader("Accept", org.springframework.http.MediaType.TEXT_EVENT_STREAM_VALUE);
//                builder.addHeader("Cache-Control", "no-cache");
//                builder.addHeader("Connection", "keep-alive");
//            }
//        }
//        switch (contextParam.getMethod()) {
//            case "PUT":
//            case "DELETE":
//            case "POST":
//                MediaType mediaType = MediaType.parse(contextParam.getMediaType());
//                omsLogger.info("request mediaType: {}", mediaType);
//                RequestBody requestBody = RequestBody.create(mediaType, JSON.toJSONString(subTask));
//                builder.method(contextParam.getMethod(), requestBody);
//                break;
//            default:
//                builder.get();
//        }
//        omsLogger.info("dispatchBusinessTask：子任务：{} 组装request body完成", subTask.getShardNum());
//
//        if (contextParam.getFluent()) {
//            return requestWithFluent(omsLogger, client, builder, contextParam.getTimeout());
//        } else {
//            return requestWithJson(omsLogger, client, builder);
//        }
//    }
//
//    private ProcessResult requestWithFluent(OmsLogger omsLogger, OkHttpClient client, Request.Builder builder, Integer timeout) {
//        omsLogger.info("requestWithFluent start");
//        // 定义SSE事件监听器
//        // 实例化EventSource，注册EventSource监听器
//        CountDownLatch latch = new CountDownLatch(1);
//        final List<ProcessResult> resultData = new ArrayList<>();
//        RealEventSource realEventSource = new RealEventSource(builder.build(), new EventSourceListener() {
//
//            @Override
//            public void onOpen(@Nonnull EventSource eventSource, @Nonnull Response response) {
//                omsLogger.info("requestWithFluent start open channel");
//            }
//
//            @Override
//            public void onEvent(@Nonnull EventSource eventSource, String id, String type, @Nonnull String data) {
//                try {
//                    omsLogger.info("requestWithFluent receive data, iD:{}, type:{}, data:{}", id, type, data);
//                    SseOutputInfo outputInfo = JSON.parseObject(data, SseOutputInfo.class);
//                    if (StringUtils.equals(outputInfo.getStage(), SseStage.PROCESSING.getCode())) {
//                        resultData.add(unstructResult(outputInfo));
//                    }
//                } catch (Exception e) {
//                    resultData.add(new ProcessResult(false, data));
//                }
//            }
//
//            private ProcessResult unstructResult(SseOutputInfo outputInfo) {
//                try {
//                    // 要求业务接口返回值是PowerJobResult，但是也兼容非PowerJobResult的结果，只要最外层有code值即可
//                    JSONObject result = JSONObject.parseObject(outputInfo.getData());
//                    if (result.containsKey(ProcessorConstant.RESULT_KEY_SUCCESS) && result.containsKey(ProcessorConstant.RESULT_KEY_data)) {
//                        return new ProcessResult(result.getBoolean(ProcessorConstant.RESULT_KEY_SUCCESS), outputInfo.getData());
//                    }
//                    return new ProcessResult(result.getInteger(ProcessorConstant.RESULT_KEY_CODE) == ProcessorConstant.HTTP_SUCCESS_CODE, outputInfo.getData());
//                } catch (Exception e) {
//                    return new ProcessResult(false, outputInfo.getData());
//                }
//            }
//
//            @Override
//            public void onClosed(@Nonnull EventSource eventSource) {
//                try {
//                    omsLogger.info("requestWithFluent end close channel");
//                    if (resultData.isEmpty()) {
//                        omsLogger.info("requestWithFluent end, but no result found, return default failed");
//                        resultData.add(new ProcessResult(false, "no data received"));
//                    }
//                    latch.countDown();
//                } catch (Exception e) {
//                    resultData.add(new ProcessResult(false, e.getMessage()));
//                } finally {
//                    latch.countDown();
//                }
//            }
//
//            @Override
//            public void onFailure(@Nonnull EventSource eventSource, Throwable t, Response response) {
//                try {
//                    omsLogger.info("requestWithFluent end failed, response:{}", JSON.toJSONString(response), t);
//                    assert response.body() != null;
//                    resultData.add(new ProcessResult(false, "http code:" + response.code() + response.body().string()));
//                } catch (Exception e) {
//                    omsLogger.info("requestWithFluent end onFailure failed:{}", JSON.toJSONString(response), t);
//                    resultData.add(new ProcessResult(false, e.getMessage()));
//                } finally {
//                    latch.countDown();
//                }
//            }
//        });
//        try {
//            realEventSource.connect(client);//真正开始请求的一步
//            if (latch.await(timeout, TimeUnit.SECONDS)) {
//                omsLogger.info("requestWithFluent end");
//                return resultData.get(resultData.size() - 1);
//            } else {
//                omsLogger.info("requestWithFluent failed job timeout in {} seconds", timeout);
//                // 断开连接
//                realEventSource.cancel();
//                return new ProcessResult(false, "job timeout");
//            }
//        } catch (Exception e) {
//            omsLogger.error("count down latch error");
//            latch.countDown();
//            return new ProcessResult(false, e.getMessage());
//        }
//    }
//
//    private static ProcessResult requestWithJson(OmsLogger omsLogger, OkHttpClient client, Request.Builder builder) throws IOException {
//        Response response = client.newCall(builder.build()).execute();
//        omsLogger.info("requestWithJson response: {}", response);
//
//        // http请求code和业务code全部成功任务才算成功，否则就是失败
//        String msgBody = "";
//        if (response.body() != null) {
//            msgBody = response.body().string();
//        }
//
//        int responseCode = response.code();
//        boolean success = false;
//        Integer businessCode = JSON.parseObject(msgBody).getInteger(ProcessorConstant.RESULT_KEY_CODE);
//        if (responseCode == ProcessorConstant.HTTP_SUCCESS_CODE && businessCode == ProcessorConstant.HTTP_SUCCESS_CODE) {
//            success = true;
//        }
//        return new ProcessResult(success, msgBody);
//    }
//
//
//    private void validateParam(OmsLogger omsLogger, HttpModel<MapReduceHttpModel> httpParams) {
////        HttpProcessor.HttpParams httpParams = JSON.parseObject(CommonUtils.parseParams(taskContext), HttpProcessor.HttpParams.class);
//        // 1、校验补全http参数
//        validHttpParam(omsLogger, httpParams);
//        // 2、校验分片参数
//        validShardingParam(omsLogger, httpParams);
//    }
//
//    private void validShardingParam(OmsLogger omsLogger, HttpModel<MapReduceHttpModel> httpParams) {
//        MapReduceHttpModel body = httpParams.getBody();
//        if (ObjectUtils.isEmpty(body) || body.getShardTotal() == null || body.getShardTotal() < 1) {
//            omsLogger.error("validShardingParam error: requestBody is empty or shardTotal not valid");
//            throw new RuntimeException("validShardingParam error: requestBody is empty or shardTotal not valid");
//        }
//        // mapreduce场景，开始时间，结束时间必传，
//        if(body.getStartTime() == null || body.getEndTime() == null) {
//            throw new RuntimeException("validShardingParam error: startTime or endTime is empty");
//        }
//    }
//
//    private static void validHttpParam(OmsLogger omsLogger, HttpModel<MapReduceHttpModel> httpParams) {
//        String instanceId = ThreadLocalStore.getLocalData(ProcessorConstant.INSTANCE_ID);
//        if (httpParams == null) {
//            String message = "httpParams is null, please check jobParam configuration.";
//            omsLogger.warn("MapHttpProcessor[{}]->validHttpParam:" + message, instanceId);
//            throw new RuntimeException("validHttpParam:httpParams is null");
//        }
//
//        if (StringUtils.isEmpty(httpParams.getUrl()) || !httpParams.getUrl().startsWith("http")) {
//            throw new RuntimeException("validHttpParam:httpParams url invalid:" + httpParams.getUrl());
//        }
//
//        omsLogger.info("MapHttpProcessor[{}]->validHttpParam: request url: {}", instanceId, httpParams.getUrl());
//
//        // set default method
//        if (StringUtils.isEmpty(httpParams.getMethod())) {
//            httpParams.setMethod(HttpMethod.GET.name());
//            omsLogger.info("MapHttpProcessor[{}]->validHttpParam:using default request method: GET", instanceId);
//        } else {
//            httpParams.setMethod(httpParams.getMethod().toUpperCase());
//            omsLogger.info("MapHttpProcessor[{}]->validHttpParam:request method: {}", instanceId);
//        }
//
//        // set default mediaType
//        if (!HttpMethod.GET.name().equals(httpParams.getMethod())) {
//            // set default request body
//            if (ObjectUtils.isEmpty(httpParams.getBody())) {
//                httpParams.setBody(new MapReduceHttpModel());
//                omsLogger.warn("MapHttpProcessor[{}]->validHttpParam:try to use default request body:{}", instanceId, httpParams.getBody());
//            }
//            if (StringUtils.isEmpty(httpParams.getMediaType())) {
//                httpParams.setMediaType(org.springframework.http.MediaType.APPLICATION_JSON.toString());
//                omsLogger.warn("MapHttpProcessor[{}]->validHttpParam:try to use 'application/json' as media type", instanceId);
//            }
//        }
//
//        // set default timeout
//        if (httpParams.getTimeout() == null || httpParams.getTimeout() > ProcessorConstant.DEFAULT_MAX_TIMEOUT) {
//            httpParams.setTimeout(ProcessorConstant.DEFAULT_TIMEOUT);
//        }
//        // set default timeout
//        if (httpParams.getFluent() == null) {
//            // 默认为非流式
//            httpParams.setFluent(Boolean.FALSE);
//        }
//    }
//
//
//    @Override
//    public ProcessResult reduce(TaskContext context, List<TaskResult> taskResults) {
//        OmsLogger omsLogger = context.getOmsLogger();
//        omsLogger.info("MapHttpProcessor[{}]->reduce result:{}", context.getInstanceId(), JSON.toJSONString(taskResults));
//        if (CollectionUtils.isEmpty(taskResults)) {
//            omsLogger.error("MapHttpProcessor[{}]->reduce result is empty", context.getInstanceId());
//        }
//        List<TaskResult> failedSubTask = taskResults.stream()
//                .filter(res -> !res.isSuccess())
//                .filter(res -> !StringUtils.equals(res.getResult(), ProcessorConstant.ROOT_TASK_RESULT))
//                .collect(Collectors.toList());
//
//        if (!CollectionUtils.isEmpty(failedSubTask)) {
//            omsLogger.error("MapHttpProcessor[{}]->reduce result exsit fail task:{}", context.getInstanceId(), JSON.toJSONString(failedSubTask));
//            return new ProcessResult(false, JSON.toJSONString(failedSubTask));
//        }
//
//        return new ProcessResult(true, "success");
//    }
//
//    @Getter
//    @NoArgsConstructor
//    @AllArgsConstructor
//    public static class SubTask {
//        private int index;
//        private String params;
//    }
//
//}
